package com.zhang.third.day06;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @title: 左外连接查询
 * @author: zhang
 * @date: 2022/4/9 16:36
 */
public class Example10 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple2<String, Long>> left = env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String value) throws Exception {
                        String[] arr = value.split(" ");
                        return Tuple2.of(arr[0], Long.parseLong(arr[1]) * 1000L);
                    }
                });

        SingleOutputStreamOperator<Tuple2<String, Long>> right = env
                .socketTextStream("localhost", 9998)
                .map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String value) throws Exception {
                        String[] arr = value.split(" ");
                        return Tuple2.of(arr[0], Long.parseLong(arr[1]) * 1000L);
                    }
                });

        left
                .keyBy(r -> r.f0)
                .connect(right.keyBy(r -> r.f0))
                .process(new CoProcessFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>() {
                    private ValueState<Tuple2<String, Long>> leftHistory;
                    private ValueState<Tuple2<String, Long>> rightHistory;
                    //定义定时器时间状态变量
                    private ValueState<Long> timerTs;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        leftHistory = getRuntimeContext().getState(
                                new ValueStateDescriptor<Tuple2<String, Long>>(
                                        "left", Types.TUPLE(Types.STRING, Types.LONG)
                                )
                        );

                        rightHistory = getRuntimeContext().getState(
                                new ValueStateDescriptor<Tuple2<String, Long>>(
                                        "right", Types.TUPLE(Types.STRING, Types.LONG)
                                )
                        );

                        timerTs = getRuntimeContext().getState(
                                new ValueStateDescriptor<Long>(
                                        "timer-Ts", Types.LONG
                                )
                        );
                    }

                    @Override
                    public void processElement1(Tuple2<String, Long> value, CoProcessFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>.Context ctx, Collector<String> out) throws Exception {
                        if (rightHistory.value() == null) {
                            // left 数据先到达等待10s right数据
                            leftHistory.update(value);
                            long currTs = ctx.timerService().currentProcessingTime();
                            long tenSecLater = currTs + 10 * 1000L;
                            ctx.timerService().registerProcessingTimeTimer(tenSecLater);
                            timerTs.update(tenSecLater);
                        } else {
                            // left数据来的时候 right 数据已经到达
                            // join成功直接输出
                            out.collect(value + "=>" + rightHistory.value());
                            // 删除right数据等待left数据时候注册的定时器
                            ctx.timerService().deleteProcessingTimeTimer(timerTs.value());
                            // 清空状态
                            rightHistory.clear();
                            timerTs.clear();
                        }

                    }

                    @Override
                    public void processElement2(Tuple2<String, Long> value, CoProcessFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>.Context ctx, Collector<String> out) throws Exception {
                        if (leftHistory.value() == null) {
                            // right 数据到达 left数据没有到达
                            // 注册定时器等待left数据 10s
                            rightHistory.update(value);
                            long currTs = ctx.timerService().currentProcessingTime();
                            long tenSecLater = currTs + 10 * 1000L;
                            ctx.timerService().registerProcessingTimeTimer(tenSecLater);
                            timerTs.update(tenSecLater);
                        } else {
                            // left 数据已经到达 join成功 直接输出
                            out.collect(leftHistory.value() + "=>" + value);
                            // 删除left等待right数据时候注册的定时器
                            ctx.timerService().deleteProcessingTimeTimer(timerTs.value());
                            timerTs.clear();
                            leftHistory.clear();
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, CoProcessFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        if (leftHistory.value() != null){
                            out.collect(leftHistory.value()+"=>"+"(null,null)");
                        }
                        leftHistory.clear();
                        rightHistory.clear();
                        timerTs.clear();
                    }
                })
                .print();

        env.execute();
    }
}
